package com.clp.protocol.iec104.client;

import com.clp.protocol.core.common.connect.NettyClient;
import com.clp.protocol.core.common.connect.RecvCallback;
import com.clp.protocol.core.common.connect.SendCallback;
import com.clp.protocol.core.exception.FailedToConnectException;
import com.clp.protocol.core.exception.FailedToSendFrameException;
import com.clp.protocol.iec104.apdu.*;
import com.clp.protocol.iec104.apdu.asdu.IAsdu;
import com.clp.protocol.iec104.client.async.*;
import com.clp.protocol.iec104.client.async.send_apdu.*;
import com.clp.protocol.iec104.client.master_config.MasterConnConfig;
import com.clp.protocol.iec104.client.master_state.*;
import com.clp.protocol.iec104.client.pipeline.Iec104ChannelInitializer;
import com.clp.protocol.iec104.connect.ConnInfoImpl;
import com.clp.protocol.iec104.connect.async.res.DefaultRes;
import com.clp.protocol.iec104.connect.async.res.SendTestFrVRes;
import com.clp.protocol.iec104.definition.Tc;
import com.clp.protocol.iec104.definition.qua_type.TaQuaType;
import com.clp.protocol.iec104.exception.IncorrectRtuAddrException;
import com.clp.protocol.iec104.exception.IncorrectSeqException;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import lombok.extern.slf4j.Slf4j;

import java.util.Collection;
import java.util.Vector;

@Slf4j
public class MasterImpl extends ConnInfoImpl implements Master {
    private final NettyClient<Iec104ChannelInitializer> nettyClient;
    private final Iec104ChannelInitializer initializer;
    private volatile boolean canReconnect = true;
    private volatile boolean isClientCloseChannel = false;

    private final int maxReconnectCount;
    private final boolean isAutoStartDtV;
    private final int totalCall100PeriodSeconds;
    private final int totalCall101PeriodSeconds;
    private final int t0Seconds;
    private final int t1Seconds;
    private final int t2Seconds;
    private final int t3Seconds;
    private final MasterStateManager masterStateManager;

    // 回调列表
    protected final Vector<RecvCallback<Apdu>> recvCallbacks = new Vector<>();
    protected final Vector<SendCallback<Apdu>> sendCallbacks = new Vector<>();

    protected MasterImpl(NettyClient<Iec104ChannelInitializer> nettyClient, MasterConnConfig cfg) {
        super(cfg, nettyClient.scheduledExecutor());
        this.nettyClient = nettyClient;
        this.maxReconnectCount = cfg.getMaxReconnectCount();
        this.isAutoStartDtV = cfg.isAutoStartDtV();
        this.totalCall100PeriodSeconds = cfg.getTotalCall100PeriodSeconds();
        this.totalCall101PeriodSeconds = cfg.getTotalCall101PeriodSeconds();
        this.t0Seconds = cfg.getT0();
        this.t1Seconds = cfg.getT1();
        this.t2Seconds = cfg.getT2();
        this.t3Seconds = cfg.getT3();
        // 回调列表
        this.initializer = new Iec104ChannelInitializer(this, cfg.getSupplier());
        this.masterStateManager = new MasterStateManager(this);
    }

    @Override
    public void resetChannel(Channel channel) {
        super.resetChannel(channel);
        masterStateManager.reset();
    }

    synchronized MasterFuture<Void> connect() {
        if (isConnected()) {
            throw new IllegalStateException("当前连接有效，无法再次进行连接！");
        }

        // 设置初始化器
        // 连接
        final ChannelFuture connFuture = nettyClient.connect(initializer, remoteHost(), remotePort());
        // 处理结果
        final MasterPromise<Void> connPromise = new DefaultMasterPromise<>(this);
        connFuture.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (connFuture.isSuccess()) {
                    resetChannel(connFuture.channel());
                    // 启动所有监视作业
                    startMonitorTasks();
                    connPromise.setSuccess();
                    log.info("连接 {}:{} 成功！", remoteHost(), remotePort());
                } else {
                    log.warn("请求连接失败，尝试重新连接");
                    reconnect0(1).addListener(new MasterFutureListener<Channel>() {
                        @Override
                        public void operationComplete(MasterFuture<Channel> future) {
                            if (future.isSuccess()) {
                                resetChannel(future.getRes());
                                startMonitorTasks();
                                connPromise.setSuccess();
                                log.info("连接 {}:{} 成功！", remoteHost(), remotePort());
                            } else {
                                connPromise.setFailure(future.cause());
                                log.error("连接 {}:{} 失败，原因：{}", remoteHost(), remotePort(), future.cause());
                            }
                        }
                    });
                    connPromise.setFailure(future.cause());
                }
            }
        });
        return connPromise.addListener(new MasterFutureListener<Void>() {
            @Override
            public void operationComplete(MasterFuture<Void> future) {
                // 如果是自动启动传输激活，那么就发送启动传输激活
                if (isAutoStartDtV()) {
                    sendUApduOfStartDtV().addListener(new MasterFutureListener<SendStartDtVRes>() {
                        @Override
                        public void operationComplete(MasterFuture<SendStartDtVRes> future) {
                            if (future.isSuccess()) {
                                log.info("自动启动传输激活成功！");
                            }
                        }
                    });
                }
            }
        });
    }

    private MasterFuture<Channel> reconnect0(int reconnCount) {
        MasterPromise<Channel> promise = new DefaultMasterPromise<>(this);
        try {
            if (reconnCount >= maxReconnectCount) {
                promise.setFailure(new FailedToConnectException("超过最大重连次数，连接失败！"));
                return promise;
            }

            log.info("进行第{}次重连", reconnCount);
            Thread.sleep(reconnCount * 1000L);

            nettyClient.connect(initializer, remoteHost(), remotePort()).addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    if (future.isSuccess()) {
                        log.info("第{}次重连成功", reconnCount);
                        promise.setRes(future.channel());
                        promise.setSuccess();
                    } else {
                        reconnect0(reconnCount + 1).addListener(new MasterFutureListener<Channel>() {
                            @Override
                            public void operationComplete(MasterFuture<Channel> future) {
                                if (future.isSuccess()) {
                                    promise.setRes(future.getRes());
                                    promise.setSuccess();
                                } else {
                                    promise.setFailure(future.cause());
                                }
                            }
                        });
                    }
                }
            });
            return promise;
        } catch (InterruptedException e) {
            promise.setFailure(e);
            return promise;
        }
    }

    public synchronized MasterFuture<Void> autoReconnect() throws CouldNotAutoReconnectException {
        if (isConnected()) {
            throw new CouldNotAutoReconnectException("自动重连只能在被动断线后进行！");
        }
        // 只能被动调用
        if (isClientCloseChannel) {
            isClientCloseChannel = false;
            throw new CouldNotAutoReconnectException("客户端主动关闭连接，此时不应自动重连！");
        }
        MasterPromise<Void> reconnPromise = new DefaultMasterPromise<>(this);
        stopMonitorTasks();
        reconnect0(1).addListener(new MasterFutureListener<Channel>() {
            @Override
            public void operationComplete(MasterFuture<Channel> future) {
                if (future.isSuccess()) {
                    resetChannel(future.getRes());
                    startMonitorTasks();
                    reconnPromise.setSuccess();
                } else {
                    reconnPromise.setFailure(future.cause());
                }
            }
        });

        return reconnPromise;
    }

    /**
     * 客户端主动重连
     *
     * @return
     */
    synchronized MasterFuture<Void> reconnect() {
        MasterPromise<Void> reconnPromise = new DefaultMasterPromise<>(this);
        stopMonitorTasks();
        if (isConnected()) {
            closeChannel(true).addListener(new MasterFutureListener<Void>() {
                @Override
                public void operationComplete(MasterFuture<Void> future) {
                    if (!future.isSuccess()) {
                        reconnPromise.setFailure(future.cause());
                        return;
                    }
                    reconnect0(1).addListener(new MasterFutureListener<Channel>() {
                        @Override
                        public void operationComplete(MasterFuture<Channel> future) {
                            if (future.isSuccess()) {
                                resetChannel(future.getRes());
                                startMonitorTasks();
                                reconnPromise.setSuccess();
                            } else {
                                reconnPromise.setFailure(future.cause());
                            }
                        }
                    });
                }
            });
        } else {
            reconnect0(1).addListener(new MasterFutureListener<Channel>() {
                @Override
                public void operationComplete(MasterFuture<Channel> future) {
                    if (future.isSuccess()) {
                        resetChannel(future.getRes());
                        startMonitorTasks();
                        reconnPromise.setSuccess();
                    } else {
                        reconnPromise.setFailure(future.cause());
                    }
                }
            });
        }

        return reconnPromise;
    }

    synchronized MasterFuture<Void> disconnect() {
        DefaultMasterPromise<Void> promise = new DefaultMasterPromise<>(this);
        stopMonitorTasks();
        if (!isConnected()) {
            log.info("连接 {}:{} 已断开！", remoteHost(), remotePort());
            promise.setSuccess();
            return promise;
        }

        return closeChannel(true).addListener(new MasterFutureListener<Void>() {
            @Override
            public void operationComplete(MasterFuture<Void> future) {
                if (future.isSuccess()) {
                    log.info("断开连接 {}:{} 成功！", remoteHost(), remotePort());
                } else {
                    log.error("断开连接 {}:{} 失败！", remoteHost(), remotePort());
                }
            }
        });
    }

    @Override
    protected boolean needResetCurrRecvCountByRecving(Apdu apdu) {
        if (!apdu.isIType()) {
            return false;
        }
        IAsdu iAsdu = apdu.castToIType().getIAsdu();
        switch (iAsdu.getTypeTag()) {
            case C_IC_NA_1: // 总召唤
                switch (iAsdu.getCot().getCause()) {
                    case COT_ACTTERM: // 激活终止
                        return true;
                }
                break;
        }
        return false;
    }

    @Override
    public boolean isAutoStartDtV() {
        return isAutoStartDtV;
    }

    @Override
    public int totalCall100PeriodSeconds() {
        return totalCall100PeriodSeconds;
    }

    @Override
    public int totalCall101PeriodSeconds() {
        return totalCall101PeriodSeconds;
    }

    @Override
    public int t0Seconds() {
        return t0Seconds;
    }

    @Override
    public int t1Seconds() {
        return t1Seconds;
    }

    @Override
    public int t2Seconds() {
        return t2Seconds;
    }

    @Override
    public int t3Seconds() {
        return t3Seconds;
    }

    @Override
    public MasterStateInfo stateInfo() {
        return masterStateManager;
    }

    protected void startMonitorTasks() {
        masterStateManager.startMonitorTask();
    }

    protected void stopMonitorTasks() {
        masterStateManager.tryStopMonitorTask();
    }

    public Apdu handleRecving(Apdu apdu) throws IncorrectRtuAddrException, IncorrectSeqException {
        // 1、检查 apdu
        checkRecvApduRtuAddr(apdu);
        checkRecvApduSeq(apdu);
        // 2、处理序号
        if (apdu.isIType()) {
            addRecvSeqCircularly();
            addCurrRecvCountCircularly();
        }

        // 3、更新主站状态
        apdu = masterStateManager.updateByRecving(apdu);
        // 4、执行回调
        handleRecvCallbacks(apdu);
        // 5、一定要返回确认帧的情况
        if (apdu.isIType()) {
            if (needResetCurrRecvCountByRecving(apdu)) {
                resetCurrRecvCount();
            }
            // 6、如果当前累计接收到达最大值，发送一个确认帧
            if (needSendSApduOfAck()) {
                log.info("达到最大累计接收I帧数");
                sendSApduOfAck();
            }
        }
        return apdu;
    }

    public Apdu handleSending(Apdu apdu) {
        // 先更新主站状态
        apdu = masterStateManager.updateBySending(apdu);
        // 再更新连接状态
        if (apdu == null) return null;
        // 发送前处理：更新序号
        if (apdu.isIType()) {
            addSendSeqCircularly();
        }
        return apdu;
    }

    @Override
    public MasterFuture<DefaultRes> sendUApduOfTestFrC() {
        DefaultMasterPromise<DefaultRes> sendPromise = new DefaultMasterPromise<>(this, new DefaultRes());
        UApdu uApdu = ApduFactory.getUApduOfTestFrC();
        send(uApdu, sendPromise).addListener((ChannelFutureListener) future -> {
            if (future.isSuccess()) {
                sendPromise.getRes().setSendSuccess(true);
                sendPromise.setSuccess();
                return;
            }
            sendPromise.getRes().setSendSuccess(false);
            sendPromise.setFailure(new FailedToSendFrameException(uApdu));
        });
        return sendPromise;
    }

    @Override
    public MasterFuture<DefaultRes> sendSApduOfAck() {
        DefaultMasterPromise<DefaultRes> sendPromise = new DefaultMasterPromise<>(this, new DefaultRes());
        SApdu sApdu = ApduFactory.getSApdu(recvSeq());
        send(sApdu, sendPromise).addListener((ChannelFutureListener) future -> {
            if (future.isSuccess()) {
                sendPromise.getRes().setSendSuccess(true);
                sendPromise.setSuccess();
                return;
            }
            sendPromise.getRes().setSendSuccess(false);
            sendPromise.setFailure(new FailedToSendFrameException(sApdu));
        });
        resetCurrRecvCount(); // 计数器重置为0
        return sendPromise;
    }

    @Override
    public MasterFuture<SendTestFrVRes> sendUApduOfTestFrV() {
        DefaultMasterPromise<SendTestFrVRes> sendPromise = new DefaultMasterPromise<>(this, new SendTestFrVRes());
        UApdu uApdu = ApduFactory.getUApduOfTestFrV();
        if (stateInfo().isTestingFr()) {
            sendPromise.getRes().setSendSuccess(false).setRecvAckSuccess(false);
            sendPromise.setFailure(new FailedToSendFrameException(uApdu));
            return sendPromise;
        }

        // 注册
        masterStateManager.getState(T0T1T2T3MasterState.class).register(sendPromise);
        // 发送
        send(uApdu, sendPromise).addListener((ChannelFutureListener) future -> {
            if (future.isSuccess()) {
                sendPromise.getRes().setSendSuccess(true);
                return;
            }
            sendPromise.getRes().setSendSuccess(false).setRecvAckSuccess(false);
            sendPromise.setFailure(new FailedToSendFrameException(uApdu));
        });
        return sendPromise;
    }

    @Override
    public MasterFuture<SendStartDtVRes> sendUApduOfStartDtV() {
        UApdu uApdu = ApduFactory.getUApduOfStartDtV();
        MasterPromise<SendStartDtVRes> sendPromise = new DefaultMasterPromise<>(this, new SendStartDtVRes());
        if (stateInfo().isStartedDt()) {
            sendPromise.getRes().setSendSuccess(false).setRecvAck(false);
            sendPromise.setFailure(new FailedToSendFrameException(uApdu));
            return sendPromise;
        }
        // 将 这个Promise注册到对应的状态中
        masterStateManager.getState(DtMasterState.class).register(sendPromise);
        send(uApdu, sendPromise);

        return sendPromise;
    }

    @Override
    public MasterFuture<SendTotalCall100Res> sendIApduOfTotalCall100() {
        IApdu iApdu = ApduFactory.getIApduOfTotalCall100(sendSeq(), recvSeq(), rtuAddr());
        MasterPromise<SendTotalCall100Res> sendPromise = new DefaultMasterPromise<>(this, new SendTotalCall100Res());
        if (!stateInfo().isRecvedInitEnd() || stateInfo().isDoingTotalCall100()) {
            sendPromise.getRes().setSendSuccess(false).setRecvTotalCall100Ack(false).setRecvTotalCall100End(false);
            sendPromise.setFailure(new FailedToSendFrameException(iApdu));
            return sendPromise;
        }
        // 将 这个Promise注册到对应的状态中
        masterStateManager.getState(TotalCall100MasterState.class).register(sendPromise);
        send(iApdu, sendPromise);
        return sendPromise;
    }

    @Override
    public MasterFuture<SendTotalCall101Res> sendIApduOfTotalCall101() {
        IApdu iApdu = ApduFactory.getIApduOfTotalCall101(sendSeq(), recvSeq(), rtuAddr());
        MasterPromise<SendTotalCall101Res> sendPromise = new DefaultMasterPromise<>(this, new SendTotalCall101Res());
        if (!stateInfo().isRecvedInitEnd() || stateInfo().isDoingTotalCall101()) {
            sendPromise.getRes().setSendSuccess(false).setRecvTotalCall101Ack(false).setRecvTotalCall101End(false);
            sendPromise.setFailure(new FailedToSendFrameException(iApdu));
            return sendPromise;
        }
        // 将 这个Promise注册到对应的状态中
        masterStateManager.getState(TotalCall101MasterState.class).register(sendPromise);
        send(iApdu, sendPromise);
        return sendPromise;
    }

    @Override
    public MasterFuture<SendTcSelectRes> sendIApduOfOnePointTcSelect(int infoObjAddr, boolean isSwitchOn) {
        Apdu iApdu = ApduFactory.getIApduOfOnePointControl(sendSeq(), recvSeq(), rtuAddr(),
                infoObjAddr, Tc.CmdType.SELECT, isSwitchOn ? Tc.OnePointSwitch.ON : Tc.OnePointSwitch.OFF);
        MasterPromise<SendTcSelectRes> sendPromise = new SendTcSelectMasterPromise(this, new SendTcSelectRes(), rtuAddr(), infoObjAddr);
        if (!stateInfo().isRecvedInitEnd() || stateInfo().isDoingTcSelect(infoObjAddr)) {
            sendPromise.getRes().setSendSelectSuccess(false).setRecvSelectAckYes(false).setRecvSelectAckNo(false).setFailDesc("该测点正在进行遥控");
            sendPromise.setFailure(new FailedToSendFrameException(iApdu));
            return sendPromise;
        }
        // 注册
        masterStateManager.getState(TcMasterState.class).register(sendPromise);
        send(iApdu, sendPromise);
        return sendPromise;
    }

    @Override
    public MasterFuture<SendTcSelectRes> sendIApduOfTwoPointTcSelect(int infoObjAddr, boolean isSwitchOn) {
        Apdu iApdu = ApduFactory.getIApduOfTwoPointControl(sendSeq(), recvSeq(), rtuAddr(),
                infoObjAddr, Tc.CmdType.SELECT, isSwitchOn ? Tc.TwoPointSwitch.ON : Tc.TwoPointSwitch.OFF);
        MasterPromise<SendTcSelectRes> sendPromise = new SendTcSelectMasterPromise(this, new SendTcSelectRes(), rtuAddr(), infoObjAddr);
        if (!stateInfo().isRecvedInitEnd() || stateInfo().isDoingTcSelect(infoObjAddr)) {
            sendPromise.getRes().setSendSelectSuccess(false).setRecvSelectAckYes(false).setRecvSelectAckNo(false).setFailDesc("该测点正在进行遥控");
            sendPromise.setFailure(new FailedToSendFrameException(iApdu));
            return sendPromise;
        }
        // 注册
        masterStateManager.getState(TcMasterState.class).register(sendPromise);
        send(iApdu, sendPromise);
        return sendPromise;
    }

    @Override
    public MasterFuture<SendTcExecuteRes> sendIApduOfOnePointTcExecute(int infoObjAddr, boolean isSwitchOn) {
        Apdu iApdu = ApduFactory.getIApduOfOnePointControl(sendSeq(), recvSeq(), rtuAddr(),
                infoObjAddr, Tc.CmdType.EXECUTE, isSwitchOn ? Tc.OnePointSwitch.ON : Tc.OnePointSwitch.OFF);
        MasterPromise<SendTcExecuteRes> sendPromise = new SendTcExecuteMasterPromise(this, new SendTcExecuteRes(), rtuAddr(), infoObjAddr);
        if (!stateInfo().isRecvedInitEnd() || stateInfo().isDoingTcExecute(infoObjAddr)) {
            sendPromise.getRes().setSendExecuteSuccess(false).setRecvExecuteAckYes(false).setRecvExecuteAckNo(false).setRecvEnd(false)
                    .setFailDesc("该测点正在进行遥控");
            sendPromise.setFailure(new FailedToSendFrameException(iApdu));
            return sendPromise;
        }
        // 注册
        masterStateManager.getState(TcMasterState.class).register(sendPromise);
        send(iApdu, sendPromise);
        return sendPromise;
    }

    @Override
    public MasterFuture<SendTcExecuteRes> sendIApduOfTwoPointTcExecute(int infoObjAddr, boolean isSwitchOn) {
        Apdu iApdu = ApduFactory.getIApduOfTwoPointControl(sendSeq(), recvSeq(), rtuAddr(),
                infoObjAddr, Tc.CmdType.EXECUTE, isSwitchOn ? Tc.TwoPointSwitch.ON : Tc.TwoPointSwitch.OFF);
        MasterPromise<SendTcExecuteRes> sendPromise = new SendTcExecuteMasterPromise(this, new SendTcExecuteRes(), rtuAddr(), infoObjAddr);
        if (!stateInfo().isRecvedInitEnd() || stateInfo().isDoingTcExecute(infoObjAddr)) {
            sendPromise.getRes().setSendExecuteSuccess(false).setRecvExecuteAckYes(false).setRecvExecuteAckNo(false).setRecvEnd(false)
                    .setFailDesc("该测点正在进行遥控");
            sendPromise.setFailure(new FailedToSendFrameException(iApdu));
            return sendPromise;
        }
        // 注册
        masterStateManager.getState(TcMasterState.class).register(sendPromise);
        send(iApdu, sendPromise);
        return sendPromise;
    }

    @Override
    public MasterFuture<SendTaSelectRes> sendIApduOfTaSelectNormalized(int infoObjAddr, int setVal) {
        Apdu iApdu = ApduFactory.getIApduOfTaNormalized(sendSeq(), recvSeq(), rtuAddr(), infoObjAddr, setVal, TaQuaType.SELECT);
        MasterPromise<SendTaSelectRes> sendPromise = new SendTaSelectMasterPromise(this, new SendTaSelectRes(), rtuAddr(), infoObjAddr);
        if (!stateInfo().isRecvedInitEnd() || stateInfo().isDoingTaSelect(infoObjAddr)) {
            sendPromise.getRes().setSendSelectSuccess(false).setRecvSelectAckYes(false).setRecvSelectAckNo(false).setFailDesc("该测点正在进行遥调");
            sendPromise.setFailure(new FailedToSendFrameException(iApdu));
        }
        // 注册
        masterStateManager.getState(TaMasterState.class).register(sendPromise);
        send(iApdu, sendPromise);
        return sendPromise;
    }

    @Override
    public MasterFuture<SendTaExecuteRes> sendIApduOfTaExecuteNormalized(int infoObjAddr, int setVal) {
        Apdu iApdu = ApduFactory.getIApduOfTaNormalized(sendSeq(), recvSeq(), rtuAddr(), infoObjAddr, setVal, TaQuaType.EXECUTE);
        MasterPromise<SendTaExecuteRes> sendPromise = new SendTaExecuteMasterPromise(this, new SendTaExecuteRes(), rtuAddr(), infoObjAddr);
        if (!stateInfo().isRecvedInitEnd() || stateInfo().isDoingTaExecute(infoObjAddr)) {
            sendPromise.getRes().setSendExecuteSuccess(false).setRecvExecuteAckYes(false).setRecvExecuteAckNo(false).setFailDesc("该测点正在进行遥调");
            sendPromise.setFailure(new FailedToSendFrameException(iApdu));
        }
        // 注册
        masterStateManager.getState(TaMasterState.class).register(sendPromise);
        send(iApdu, sendPromise);
        return sendPromise;
    }

    @Override
    public MasterFuture<SendTaSelectRes> sendIApduOfTaSelectFloat(int infoObjAddr, float setVal) {
        Apdu iApdu = ApduFactory.getIApduOfTaFloat(sendSeq(), recvSeq(), rtuAddr(), infoObjAddr, setVal, TaQuaType.SELECT);
        MasterPromise<SendTaSelectRes> sendPromise = new SendTaSelectMasterPromise(this, new SendTaSelectRes(), rtuAddr(), infoObjAddr);
        if (!stateInfo().isRecvedInitEnd() || stateInfo().isDoingTaSelect(infoObjAddr)) {
            sendPromise.getRes().setSendSelectSuccess(false).setRecvSelectAckYes(false).setRecvSelectAckNo(false).setFailDesc("该测点正在进行遥调");
            sendPromise.setFailure(new FailedToSendFrameException(iApdu));
        }
        // 注册
        masterStateManager.getState(TaMasterState.class).register(sendPromise);
        send(iApdu, sendPromise);
        return sendPromise;
    }

    @Override
    public MasterFuture<SendTaExecuteRes> sendIApduOfTaExecuteFloat(int infoObjAddr, float setVal) {
        Apdu iApdu = ApduFactory.getIApduOfTaFloat(sendSeq(), recvSeq(), rtuAddr(), infoObjAddr, setVal, TaQuaType.EXECUTE);
        MasterPromise<SendTaExecuteRes> sendPromise = new SendTaExecuteMasterPromise(this, new SendTaExecuteRes(), rtuAddr(), infoObjAddr);
        if (!stateInfo().isRecvedInitEnd() || stateInfo().isDoingTaExecute(infoObjAddr)) {
            sendPromise.getRes().setSendExecuteSuccess(false).setRecvExecuteAckYes(false).setRecvExecuteAckNo(false).setFailDesc("该测点正在进行遥调");
            sendPromise.setFailure(new FailedToSendFrameException(iApdu));
        }
        // 注册
        masterStateManager.getState(TaMasterState.class).register(sendPromise);
        send(iApdu, sendPromise);
        return sendPromise;
    }

    private <T> ChannelFuture send(Apdu apdu, MasterPromise<T> sendPromise) {
        sendCallbacks.forEach(apduSendCallback -> apduSendCallback.beforeSend(apdu));
        sendPromise.addListener(new MasterFutureListener<T>() {
            @Override
            public void operationComplete(MasterFuture<T> future) {
                if (future.isSuccess()) {
                    sendCallbacks.forEach(apduSendCallback -> apduSendCallback.afterSendSuccess(apdu));
                } else {
                    sendCallbacks.forEach(apduSendCallback -> apduSendCallback.afterSendFailed(apdu, future.cause()));
                }
            }
        });
        return channel().writeAndFlush(apdu).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (!future.isSuccess()) {
                    sendPromise.setFailure(future.cause());
                }
            }
        });
    }

    @Override
    public void addSendCallback(SendCallback<Apdu> callback) {
        sendCallbacks.add(callback);
    }

    @Override
    public void addSendCallbacks(Collection<SendCallback<Apdu>> sendCallbacks) {
        this.sendCallbacks.addAll(sendCallbacks);
    }

    @Override
    public void removeSendCallback(SendCallback<Apdu> callback) {
        sendCallbacks.remove(callback);
    }

    @Override
    public void removeSendCallbacks(Collection<SendCallback<Apdu>> sendCallbacks) {
        this.sendCallbacks.removeAll(sendCallbacks);
    }

    @Override
    public void addRecvCallback(RecvCallback<Apdu> callback) {
        this.recvCallbacks.add(callback);
    }

    @Override
    public void addRecvCallbacks(Collection<RecvCallback<Apdu>> recvCallbacks) {
        this.recvCallbacks.addAll(recvCallbacks);
    }

    @Override
    public void removeRecvCallback(RecvCallback<Apdu> callback) {
        this.recvCallbacks.remove(callback);
    }

    @Override
    public void removeRecvCallbacks(Collection<RecvCallback<Apdu>> recvCallbacks) {
        this.recvCallbacks.removeAll(recvCallbacks);
    }

    private void handleRecvCallbacks(Apdu apdu) {
        scheduledExecutor().submit(() -> {
            this.recvCallbacks.removeIf(callback -> callback.whenRecv(apdu));
        });
    }

    /**
     * 关闭通道
     */
    public MasterFuture<Void> closeChannel(boolean willReconnect) {
        if (!isConnected()) return new SucceedMasterPromise<>(this);
        this.isClientCloseChannel = !willReconnect;
        // 关闭通道
        DefaultMasterPromise<Void> closePromise = new DefaultMasterPromise<>(this);
        try {
            channel().close().addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    if (!future.isSuccess()) {
                        closePromise.setFailure(future.cause());
                        return;
                    }
                    closePromise.setSuccess();
                }
            });
        } catch (Exception ex) {
            closePromise.setFailure(ex.getCause());
            return closePromise;
        }
        return closePromise;
    }

    @Override
    public boolean equals(Object o) {
        return super.equals(o);
    }

    @Override
    public int hashCode() {
        return super.hashCode();
    }

    @Override
    public String toString() {
        return "Master{" +
                "remoteHost='" + remoteHost() + '\'' +
                ", remotePort=" + remotePort() +
                ", rtuAddr=" + rtuAddr() +
                '}';
    }
}
